Low-Level API

from langgraph.constants import START, END
from langgraph.graph import StateGraph
builder = StateGraph(dict)

def addition(state):
    """
    执行加法运算的节点函数

    参数:
        state (dict): 包含输入数据的状态字典，必须包含键"x"

    返回:
        dict: 返回更新后的状态字典，其中"x"的值增加1
    """
    print(f'加法节点收到的初始值:{state}')
    return {"x": state["x"] + 1}

def subtraction(state):
    """
    执行减法运算的节点函数

    参数:
        state (dict): 包含输入数据的状态字典，必须包含键"x"

    返回:
        dict: 返回更新后的状态字典，其中"x"的值减少2
    """
    print(f'减法节点收到的初始值:{state}')
    return {"x": state["x"] - 2}

# 向图构建器中添加节点
# 添加加法运算节点和减法运算节点到构建器中
builder.add_node("addition", addition)
builder.add_node("subtraction", subtraction)

# 定义节点之间的执行顺序 edges
# 设置节点间的依赖关系，形成执行流程图
builder.add_edge(START, "addition")
builder.add_edge("addition", "subtraction")
builder.add_edge("subtraction", END)
# 编译图构建器生成计算图
graph = builder.compile()

# 打印图的边和节点信息
print(builder.edges)
print(builder.nodes)
# 打印图的可视化结构
print(graph.get_graph().print_ascii())

from langgraph.constants import START, END
from langgraph.graph import StateGraph
from pydantic import BaseModel

class CalcState(BaseModel):
    """
    定义计算过程中使用的状态模型

    属性:
        x (int): 用于传递和更新的整型数值
    """
    x: int

builder = StateGraph(CalcState)

def addition(state):
    """
    执行加法运算的节点函数

    参数:
        state (CalcState): 包含输入数据的状态对象，必须包含属性"x"

    返回:
        CalcState: 返回更新后的状态对象，其中"x"的值增加1
    """
    print(f'加法节点收到的初始值:{state}')
    return CalcState(x=state.x + 1)

def subtraction(state):
    """
    执行减法运算的节点函数

    参数:
        state (CalcState): 包含输入数据的状态对象，必须包含属性"x"

    返回:
        CalcState: 返回更新后的状态对象，其中"x"的值减少2
    """
    print(f'减法节点收到的初始值:{state}')
    return CalcState(x=state.x - 2)

# 向图构建器中添加节点
# 添加加法运算节点和减法运算节点到构建器中
builder.add_node("addition", addition)
builder.add_node("subtraction", subtraction)

# 定义节点之间的执行顺序 edges
# 设置节点间的依赖关系，形成执行流程图
builder.add_edge(START, "addition")
builder.add_edge("addition", "subtraction")
builder.add_edge("subtraction", END)

# 编译图构建器生成计算图
graph = builder.compile()

# 打印图的边和节点信息
print(builder.edges)
print(builder.nodes)

# 打印图的可视化结构
print(graph.get_graph().print_ascii())

# 定义一个初始状态对象，包含属性"x"为5
initial_state = CalcState(x=5)

# 调用graph对象的invoke方法，传入初始状态，执行图计算流程
result = graph.invoke(initial_state)

print(f"最后的结果是:{result}")

from typing import Optional
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from loguru import logger
from pydantic import BaseModel

class MyState(BaseModel):
    """
    定义状态模型，用于在图节点之间传递数据

    Attributes:
        x (int): 输入的整数
        result (Optional[str]): 处理结果，可为"even"或"odd"
    """
    x: int
    result: Optional[str] = None

builder = StateGraph(MyState)

def check_x(state: MyState) -> MyState:
    """
    检查输入状态的节点函数

    Args:
        state (MyState): 包含输入数据的状态对象

    Returns:
        MyState: 返回原始状态对象，未做修改
    """
    logger.info(f"[check_x] Received state: {state}")
    return state

def is_even(state: MyState) -> bool:
    """
    判断状态中x值是否为偶数的条件函数

    Args:
        state (MyState): 包含待判断数值的状态对象

    Returns:
        bool: 如果x是偶数返回True，否则返回False
    """
    return state.x % 2 == 0

def handle_even(state: MyState) -> MyState:
    """
    处理偶数情况的节点函数

    Args:
        state (MyState): 包含偶数输入的状态对象

    Returns:
        MyState: 返回更新后的状态对象，result设置为"even"
    """
    logger.info("[handle_even] x 是偶数")
    return MyState(x=state.x, result="even")

def handle_odd(state: MyState) -> MyState:
    """
    处理奇数情况的节点函数

    Args:
        state (MyState): 包含奇数输入的状态对象

    Returns:
        MyState: 返回更新后的状态对象，result设置为"odd"
    """
    logger.info("[handle_odd] x 是奇数")
    return MyState(x=state.x, result="odd")

builder.add_node("check_x", check_x)
builder.add_node("handle_even", handle_even)
builder.add_node("handle_odd", handle_odd)

def is_even(state: MyState) -> bool:
    """
    判断状态中x值是否为偶数的条件函数

    Args:
        state (MyState): 包含待判断数值的状态对象

    Returns:
        bool: 如果x是偶数返回True，否则返回False
    """
    return state.x % 2 == 0

# 添加条件边，根据is_even函数的返回值决定流向哪个节点
builder.add_conditional_edges("check_x", is_even, {
    True: "handle_even",
    False: "handle_odd"
})

# 添加起始边，从START节点流向check_x节点
builder.add_edge(START, "check_x")

# 添加结束边，从处理节点流向END节点
builder.add_edge("handle_even", END)
builder.add_edge("handle_odd", END)

# 编译图结构
graph = builder.compile()

# 打印图结构
graph.get_graph().draw_png('./graph.png')

# 测试用例：输入偶数4
logger.info("输入 x=4(偶数)")
graph.invoke(MyState(x=4))

# 测试用例：输入奇数3
logger.info("输入 x=3(奇数)")
graph.invoke(MyState(x=3))

from langgraph.constants import START, END
from langgraph.graph import StateGraph
from loguru import logger
from pydantic import BaseModel

class LoopState(BaseModel):
    """
    循环状态模型类

    Attributes:
        x (int): 状态变量，用于循环计数
    """
    x: int

builder = StateGraph(LoopState)

def increment(state: LoopState) -> LoopState:
    """
    增量函数，将状态中的x值加1

    Args:
        state (LoopState): 包含当前x值的循环状态对象

    Returns:
        LoopState: 返回更新后的循环状态对象，其中x值增加1
    """
    logger.info(f"[increment] 当前 x = {state.x}")
    return LoopState(x=state.x + 1)

builder.add_node("increment", increment)

def is_done(state: LoopState) -> bool:
    """
    判断循环是否结束的条件函数

    Args:
        state (LoopState): 包含当前x值的循环状态对象

    Returns:
        bool: 当x值大于10时返回True，否则返回False
    """
    return state.x > 10

builder.add_conditional_edges("increment", is_done, {
    True: END,
    False: "increment"
})

builder.add_edge(START, "increment")

# 编译图结构
graph = builder.compile()

# 打印图结构
graph.get_graph().draw_png('./graph.png')

# 初始化循环并执行，直到满足结束条件
logger.info("执行循环直到 x > 10，初始x = 6")
final_state = graph.invoke(LoopState(x=6))
logger.info(f"[最终结果] -> x = {final_state['x']}")

from loguru import logger
from pydantic import BaseModel
from typing import Optional
from langgraph.graph import StateGraph, START, END

class BranchLoopState(BaseModel):
    """
    状态模型，用于保存当前流程中的变量状态。

    属性:
        x (int): 当前数值。
        done (Optional[bool]): 标记流程是否已完成，默认为 False。
    """
    x: int
    done: Optional[bool] = False

def check_x(state: BranchLoopState) -> BranchLoopState:
    """
    打印当前状态中 x 的值，用于调试和跟踪流程执行。

    参数:
        state (BranchLoopState): 包含当前 x 值的状态对象。

    返回:
        BranchLoopState: 返回未修改的原始状态对象。
    """
    logger.info(f"[check_x] 当前 x = {state.x}")
    return state

def is_even(state: BranchLoopState) -> bool:
    """
    判断当前状态中的 x 是否为偶数。

    参数:
        state (BranchLoopState): 包含当前 x 值的状态对象。

    返回:
        bool: 如果 x 是偶数则返回 True，否则返回 False。
    """
    return state.x % 2 == 0

def increment(state: BranchLoopState) -> BranchLoopState:
    """
    将当前状态中的 x 加一，并记录日志。

    参数:
        state (BranchLoopState): 包含当前 x 值的状态对象。

    返回:
        BranchLoopState: 返回更新后的状态对象(x+1)。
    """
    logger.info(f"[increment] x 是偶数，执行 +1 → {state.x + 1}")
    return BranchLoopState(x=state.x + 1)

def done(state: BranchLoopState) -> BranchLoopState:
    """
    标记流程完成，并记录日志。

    参数:
        state (BranchLoopState): 包含当前 x 值的状态对象。

    返回:
        BranchLoopState: 返回标记为完成的状态对象。
    """
    logger.info(f"[done] x 是奇数，流程结束")
    return BranchLoopState(x=state.x, done=True)

# 创建状态图并定义节点与边的关系
builder = StateGraph(BranchLoopState)
builder.add_node("check_x", check_x)
builder.add_node("increment", increment)
builder.add_node("done_node", done)

# 添加条件边：根据 is_even 函数的结果决定走向 increment 或 done_node
builder.add_conditional_edges("check_x", is_even, {
    True: "increment", False: "done_node"
})

# 定义流程路径：increment 节点之后回到 check_x 形成循环
builder.add_edge("increment", "check_x")

# 设置起始和结束节点连接
builder.add_edge(START, "check_x")
builder.add_edge("done_node", END)

# 编译状态图
graph = builder.compile()

# 绘制流程图为 PNG 图片
graph.get_graph().draw_png('./graph.png')

# 测试用例1：从偶数开始，进入循环直到变为奇数
logger.info("初始 x=6(偶数，进入循环)")
final_state1 = graph.invoke(BranchLoopState(x=6))
logger.info("[最终结果1] ->", final_state1)

# 测试用例2：从奇数开始，直接结束流程
logger.info("初始 x=3(奇数，直接 done)")
final_state2 = graph.invoke(BranchLoopState(x=3))
logger.info("[最终结果2] ->", final_state2)

from operator import add
from typing import TypedDict, Annotated
from langgraph.constants import END
from langgraph.graph import StateGraph, MessagesState, START

class State(TypedDict):
    """
    定义状态类，用于存储图节点间传递的消息状态
    messages: 使用add函数合并的字符串列表消息
    """
    messages: Annotated[list[str], add]

def sub_node(state:State) -> MessagesState:
    # 子图节点处理函数，接收当前状态并返回响应消息
    # @param state 当前状态对象，包含消息列表
    # @return 包含子图响应消息的新状态
    return {"messages": ["response from subgraph"]}

# 创建子图构建器并配置节点和边
subgraph_builder = StateGraph(State)
subgraph_builder.add_node("sub_node", sub_node)
subgraph_builder.add_edge(START, "sub_node")
subgraph_builder.add_edge("sub_node", END)
subgraph = subgraph_builder.compile()

# 绘制子图结构图
subgraph.get_graph().draw_png('./subgraph.png')

# 创建主图构建器并添加子图节点
builder = StateGraph(State)
builder.add_node("subgraph_node", subgraph)
builder.add_edge(START, "subgraph_node")
builder.add_edge("subgraph_node", END)

# 编译主图并绘制结构图
graph = builder.compile()
graph.get_graph().draw_png('./graph.png')

# 执行图并打印结果
print(graph.invoke({"messages": ["hello subgraph"]}))

from typing import TypedDict, Annotated,Literal
from langchain_ollama import ChatOllama
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph
from langgraph.types import interrupt, Command

# 定义 Agent 的状态结构，包含消息列表
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# 初始化本地大语言模型，配置模型名称和推理模式
llm = ChatOllama(base_url="http://localhost:11434", model="qwen3:8b", reasoning=False)

# 聊天机器人函数，用于处理对话状态并生成回复
def chatbot(state: AgentState):
    return {"messages": [llm.invoke(state['messages'])]}

def human_approval(state: AgentState) -> Command[Literal["chatbot", END]]:
    question = "是否同意调用大语言模型？(y/n): "
    while True:
        response = input(question).strip().lower()
        if response in ("y", "yes"):
            return Command(goto="chatbot")
        elif response in ("n", "no"):
            print("❌ 已拒绝，流程结束。")
            return Command(goto=END)
        else:
            print("⚠️ 请输入 y 或 n。")

# 构建状态图结构
graph_builder = StateGraph(AgentState)

# 每个节点都与对应的处理函数进行绑定，构成工作流的基本单元
graph_builder.add_node("human_approval", human_approval)
graph_builder.add_node("chatbot", chatbot)

# 添加边：从 START 到 chatbot，然后到 END
graph_builder.add_edge(START, "human_approval")

checkpointer=InMemorySaver()
# 编译图结构，并绘制可视化图表
graph = graph_builder.compile(checkpointer=checkpointer)
graph.get_graph().draw_png('./graph.png')
config = {"configurable": {"thread_id": "chat-1"}}
response1 = graph.invoke({"messages": ["北京天气怎么样"]},config)
print(response1["messages"][-1].content)
# 确认执行
final_result = graph.invoke(Command(resume=True),config)
print(final_result["messages"][-1].content)
# 取消执行
# final_result = graph.invoke(Command(resume=False),config)
# print(final_result["messages"][-1].content)

import json
import os
import httpx
import dotenv
from loguru import logger
from pydantic import Field, BaseModel
from langchain_core.tools import tool

# 加载环境变量配置
dotenv.load_dotenv()

class WeatherQuery(BaseModel):
    """
    天气查询参数模型类，用于定义天气查询工具的输入参数结构。

    :param city: 城市名称，字符串类型，表示要查询天气的城市
    """
    city: str = Field(description="城市名称")

class WriteQuery(BaseModel):
    """
    写入查询模型类

    用于定义需要写入文档的内容结构，继承自BaseModel基类

    属性:
        content (str): 需要写入文档的具体内容，包含详细的描述信息
    """
    content: str = Field(description="需要写入文档的具体内容")

@tool(args_schema=WeatherQuery)
def get_weather(city):
    """
    查询指定城市的即时天气信息。

    :param city: 必要参数，字符串类型，表示要查询天气的城市名称。
                 注意：中国城市需使用其英文名称，如 "Beijing" 表示北京。
    :return: 返回 OpenWeather API 的响应结果，URL 为
             https://api.openweathermap.org/data/2.5/weather。
             响应内容为 JSON 格式的字符串，包含详细的天气数据。
    """
    # 构建请求 URL
    url = "https://api.openweathermap.org/data/2.5/weather"

    # 设置查询参数
    params = {
        "q": city,  # 城市名称
        "appid": os.getenv("OPENWEATHER_API_KEY"),  # 从环境变量中读取 API Key
        "units": "metric",  # 使用摄氏度作为温度单位
        "lang": "zh_cn"  # 返回简体中文的天气描述
    }

    # 发送 GET 请求并获取响应
    response = httpx.get(url, params=params)

    # 将响应解析为 JSON 并序列化为字符串返回
    data = response.json()
    logger.info(f"查询天气结果：{json.dumps(data)}")
    return json.dumps(data)

@tool(args_schema=WriteQuery)
def write_file(content):
    """
    将指定内容写入本地文件

    参数:
        content (str): 要写入文件的文本内容

    返回值:
        str: 表示写入操作成功完成的提示信息
    """
    # 将内容写入res.txt文件，使用utf-8编码确保中文字符正确保存
    with open('res.txt', 'w', encoding='utf-8') as f:
        f.write(content)
        logger.info(f"已成功写入本地文件，写入内容：{content}")
        return "已成功写入本地文件。"

from typing import TypedDict, Annotated
from langchain_core.messages import SystemMessage
from langchain_ollama import ChatOllama
from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph
from langgraph.prebuilt import ToolNode

from tools import get_weather, write_file

# 定义 Agent 的状态结构，包含消息列表
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# 初始化本地大语言模型，配置模型名称和推理模式
llm = ChatOllama(model="qwen3:14b", reasoning=False)
tools = [get_weather, write_file]
llm_with_tools = llm.bind_tools(tools)

# 聊天机器人节点，用于处理对话状态并生成回复，并告诉模型可以调用哪些工具
def chat_node(state: AgentState):
    messages = state["messages"]
    system_prompt = """你是一个智能助手，具备以下能力：
                    1. 查询天气信息
                    2. 结果写入文件
                    请根据用户的需求，选择合适的工具来完成任务。回答要准确、友好、专业。"""
    # 构建完整的消息列表(系统提示词 + 用户消息),如果第一条消息不是系统消息，则添加系统提示词
    if not any(isinstance(msg, SystemMessage) for msg in messages):
        messages = [SystemMessage(
            content=system_prompt)] + messages
    result = llm_with_tools.invoke(messages)
    return {"messages": [result]}

# 定义工具节点(系统预置 ToolNode 会自动解析 tool_calls)
tool_node = ToolNode(tools=tools)

# 动态路由：chat_node → tool_node 或 END
def route_after_chat(state: AgentState):
    """判断是否需要进入工具节点"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tool_node"
    return END

# 构建状态图结构
graph_builder = StateGraph(AgentState)

# 每个节点都与对应的处理函数进行绑定，构成工作流的基本单元
graph_builder.add_node("chat_node", chat_node)
graph_builder.add_node("tool_node", tool_node)

# 添加边：从 START 到 chatbot，然后到 END
graph_builder.add_edge(START, "chat_node")
# 添加条件边：根据是否有工具调用来判断是否需要进入工具节点
graph_builder.add_conditional_edges("chat_node", route_after_chat, ["tool_node", END])
# 工具节点执行完后回到 chat_node，继续多轮对话
graph_builder.add_edge("tool_node", "chat_node")

# 编译图结构，并绘制可视化图表
graph = graph_builder.compile()
graph.get_graph().draw_png('./graph.png')

response1 = graph.invoke({"messages": ["北京天气怎么样"]})

print(response1["messages"][-1].content)

from typing import TypedDict, Annotated
import json
from langchain_core.messages import SystemMessage, ToolMessage
from langchain_ollama import ChatOllama
from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from tools import get_weather, write_file

# 定义 Agent 的状态结构
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# 初始化本地大语言模型
llm = ChatOllama(model="qwen3:8b", reasoning=False)
tools = [get_weather, write_file]
llm_with_tools = llm.bind_tools(tools)

# 聊天机器人节点
def chat_node(state: AgentState):
    messages = state["messages"]
    system_prompt = """你是一个智能助手，具备以下能力：
                    1. 查询天气信息
                    2. 结果写入文件
                    请根据用户的需求，选择合适的工具来完成任务。回答要准确、友好、专业。"""

    if not any(isinstance(msg, SystemMessage) for msg in messages):
        messages = [SystemMessage(content=system_prompt)] + messages

    result = llm_with_tools.invoke(messages)
    return {"messages": [result]}

# 定义工具节点
tool_node = ToolNode(tools=tools)

# 动态路由：chat_node 之后
def route_after_chat(state: AgentState):
    """判断是否需要调用工具"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tool_node"
    return END

# 构建状态图
graph_builder = StateGraph(AgentState)

# 添加节点
graph_builder.add_node("chat_node", chat_node)
graph_builder.add_node("tool_node", tool_node)

# 添加边
graph_builder.add_edge(START, "chat_node")
graph_builder.add_conditional_edges("chat_node", route_after_chat, ["tool_node", END])
graph_builder.add_edge("tool_node", "chat_node")

# 编译图结构 - 关键：使用 interrupt_before 在工具节点前中断
memory = MemorySaver()
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["tool_node"]  # 在执行工具前中断，等待人工确认
)

# 绘制可视化图表
graph.get_graph().draw_png('./graph.png')

def run_with_approval():
    """运行带人工确认的工作流"""
    config = {"configurable": {"thread_id": "1"}}

    # 第一步：发送用户消息，执行到中断点
    print("【用户】北京天气怎么样\n")
    result = graph.invoke({"messages": ["北京天气怎么样"]}, config)

    # 检查是否中断(等待人工确认)
    snapshot = graph.get_state(config)

    if snapshot.next:  # 如果有下一个节点，说明被中断了
        print("工具调用需要人工确认")

        # 获取待执行的工具调用信息
        last_message = snapshot.values["messages"][-1]
        if hasattr(last_message, "tool_calls") and last_message.tool_calls:
            for idx, tool_call in enumerate(last_message.tool_calls, 1):
                print(f"\n[{idx}] 工具名称: {tool_call['name']}")
                print(f"    调用参数: {json.dumps(tool_call['args'], ensure_ascii=False, indent=4)}")

            approval = input("是否批准执行？(yes/no): ").strip().lower()

            if approval in ['yes', 'y']:
                print("工具调用已批准，继续执行...\n")
                # 继续执行(resume)
                result = graph.invoke(None, config)

            elif approval in ['no', 'n']:
                print("工具调用已被拒绝\n")
                # 手动添加拒绝消息，然后继续
                tool_messages = []
                for tool_call in last_message.tool_calls:
                    tool_messages.append(
                        ToolMessage(
                            content="工具调用被用户拒绝，请询问用户是否需要调整方案或提供更多信息。",
                            tool_call_id=tool_call["id"]
                        )
                    )
                # 更新状态并跳过工具节点
                graph.update_state(config, {"messages": tool_messages})
                result = graph.invoke(None, config)

    # 输出最终结果
    print("最终回复:")
    final_message = result["messages"][-1]
    print(final_message.content if hasattr(final_message, 'content') else str(final_message))

# 测试运行
if __name__ == "__main__":
    run_with_approval()

from typing import TypedDict, Annotated
from langchain_core.messages import SystemMessage, AIMessage
from langchain_ollama import ChatOllama
from langgraph.checkpoint.memory import MemorySaver
from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph
from langgraph.prebuilt import ToolNode
from tools import get_weather

# 定义 Agent 的状态结构，包含消息列表和审核状态
class AgentState(TypedDict):
    """
    描述 Agent 当前状态的数据结构。

    属性:
        messages (Annotated[list, add_messages]): 包含历史交互信息的消息列表，
                                                  使用 `add_messages` 合并新旧消息。
        user_approved (bool): 标记用户是否同意执行工具调用
    """
    messages: Annotated[list, add_messages]
    user_approved: bool

# 初始化本地大语言模型，配置基础URL、模型名称和推理模式
llm = ChatOllama(base_url="http://localhost:11434", model="qwen3:14b", reasoning=False)
tools = [get_weather]
model = llm.bind_tools(tools)

def call_model(state: AgentState):
    """
    调用绑定工具的大语言模型以生成响应。

    参数:
        state (AgentState): 包含当前会话中所有消息的状态对象。

    返回:
        dict: 新增模型响应后的更新状态(仅追加最新一条回复)。
    """
    system_prompt = SystemMessage("你是一个AI助手，可以依据用户提问产生回答，你还具备调用天气函数的能力")
    response = model.invoke([system_prompt] + state["messages"])
    return {"messages": [response]}

# --- 人在闭环 (HITL) 节点 ---
def human_review(state: AgentState):
    """
    在执行工具调用之前请求人工审核确认。

    如果最后一条消息包含待执行的工具调用，则提示用户进行确认。
    若用户拒绝，则终止流程；否则允许进入工具执行阶段。

    参数:
        state (AgentState): 包含当前会话状态的对象。

    返回:
        dict: 根据用户选择决定下一步操作：
              - 用户拒绝时返回系统提示消息并标记为未批准；
              - 允许继续则标记为已批准。
    """
    last_message = state["messages"][-1]

    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        call = last_message.tool_calls[0]
        tool_name = call["name"]
        tool_args = call["args"]

        print(f"[HITL] 模型计划调用工具 `{tool_name}`，参数：{tool_args}")
        confirm = input("[HITL] 是否确认执行？(y/n): ")

        if confirm.lower() != "y":
            # 用户拒绝，返回提示消息并标记为未批准
            return {
                "messages": [AIMessage(content="用户拒绝了工具调用，无法获取相关信息。")],
                "user_approved": False
            }

        # 用户同意，标记为已批准
        return {"user_approved": True}

    # 没有工具调用，直接标记为已批准
    return {"user_approved": True}

def should_review(state: AgentState):
    """
    判断是否需要进行人工审核。

    参数:
        state (AgentState): 当前状态

    返回:
        str: 下一个节点名称
    """
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "human_review"
    return END

def should_execute_tools(state: AgentState):
    """
    判断是否应该执行工具。

    参数:
        state (AgentState): 当前状态

    返回:
        str: 下一个节点名称
    """
    if state.get("user_approved", False):
        return "tools"
    return END

# 创建工具节点，用于执行工具调用
tool_node = ToolNode(tools)

# 构建状态图结构
graph_builder = StateGraph(AgentState)

# 每个节点都与对应的处理函数进行绑定，构成工作流的基本单元
graph_builder.add_node("agent", call_model)
graph_builder.add_node("human_review", human_review)
graph_builder.add_node("tools", tool_node)

# 添加边：从 START 到 agent
graph_builder.add_edge(START, "agent")

# 添加条件边：根据是否有工具调用来判断是否需要人工审核
graph_builder.add_conditional_edges(
    "agent",
    should_review,
    {"human_review": "human_review", END: END}
)

# 添加条件边：根据用户是否同意来决定是否执行工具或结束
graph_builder.add_conditional_edges(
    "human_review",
    should_execute_tools,
    {"tools": "tools", END: END}
)

# 工具执行完成后重新回到 agent 继续对话循环
graph_builder.add_edge("tools", "agent")

# 创建内存保存器
memory = MemorySaver()

# 编译图结构，并绘制可视化图表
graph = graph_builder.compile(checkpointer=memory)
graph.get_graph().draw_png('./graph.png')

# 配置对话线程ID
config = {"configurable": {"thread_id": "chat-1"}}

# 运行第一轮：问北京天气
print("\n" + "=" * 50)
print("第一轮对话：询问北京天气")
print("=" * 50)
response1 = graph.invoke({"messages": ["北京天气怎么样"]}, config=config)

print("\n=== 第一次结果 ===")
print(response1["messages"][-1].content)

# 打印已保存的检查点
print("\n" + "=" * 50)
print("检查点历史")
print("=" * 50)
states = list(graph.get_state_history(config))

for i, state in enumerate(states):
    print(f"\n=== 检查点 {i} (next: {state.next}) ===")
    print(f"Checkpoint ID: {state.config['configurable']['checkpoint_id']}")
    if state.values.get("messages"):
        print(f"Messages count: {len(state.values['messages'])}")

# 从第二个检查点恢复并注入新问题
print("\n" + "=" * 50)
print("第二轮对话：从检查点恢复并询问上海天气")
print("=" * 50)
new_config = graph.update_state(
    states[1].config,
    values={"messages": [{"role": "user", "content": "上海天气怎么样"}]}
)

response2 = graph.invoke(None, config=new_config)

print("\n=== 第二次结果 ===")
print(response2["messages"][-1].content)

==================================================
第一轮对话：询问北京天气
==================================================
[HITL] 模型计划调用工具 `get_weather`，参数：{'city': 'Beijing'}
[HITL] 是否确认执行？(y/n): y
2025-10-05 19:05:19.651 | INFO     | tools:get_weather:61 - 查询天气结果：{"coord": {"lon": 116.3972, "lat": 39.9075}, "weather": [{"id": 804, "main": "Clouds", "description": "\u9634\uff0c\u591a\u4e91", "icon": "04n"}], "base": "stations", "main": {"temp": 17.38, "feels_like": 16.01, "temp_min": 17.38, "temp_max": 17.38, "pressure": 1019, "humidity": 32, "sea_level": 1019, "grnd_level": 1014}, "visibility": 10000, "wind": {"speed": 2.49, "deg": 140, "gust": 4.15}, "clouds": {"all": 100}, "dt": 1759662186, "sys": {"country": "CN", "sunrise": 1759616047, "sunset": 1759657887}, "timezone": 28800, "id": 1816670, "name": "Beijing", "cod": 200}

=== 第一次结果 ===
北京当前天气为多云，温度为17.38°C，体感温度为16.01°C。风速为2.49 m/s，湿度为32%。整体天气状况较为舒适。

==================================================
检查点历史
==================================================

=== 检查点 0 (next: ()) ===
Checkpoint ID: 1f0a1db2-f468-6602-8004-fde620773f92
Messages count: 4

=== 检查点 1 (next: ('agent',)) ===
Checkpoint ID: 1f0a1db2-e616-6e9c-8003-5dce54983b16
Messages count: 3

=== 检查点 2 (next: ('tools',)) ===
Checkpoint ID: 1f0a1db2-dfab-6beb-8002-022fa30c6f85
Messages count: 2

=== 检查点 3 (next: ('human_review',)) ===
Checkpoint ID: 1f0a1db2-c2dd-64c8-8001-25523b18653a
Messages count: 2

=== 检查点 4 (next: ('agent',)) ===
Checkpoint ID: 1f0a1db2-bd6d-6051-8000-d31656b995c0
Messages count: 1

=== 检查点 5 (next: ('__start__',)) ===
Checkpoint ID: 1f0a1db2-bd6b-6c22-bfff-655ead490034

==================================================
第二轮对话：从检查点恢复并询问上海天气
==================================================
[HITL] 模型计划调用工具 `get_weather`，参数：{'city': 'Shanghai'}
[HITL] 是否确认执行？(y/n): n

=== 第二次结果 ===
用户拒绝了工具调用，无法获取相关信息。